package com.codejiwei.flink.table;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * author: codejiwei
 * date: 2023/8/8
 * desc: flink table api stream2table
 **/
public class Flink_Table_API_Stream2Table {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        Person tom = new Person("Tom", 23);
        Person jerry = new Person("Jerry", 33);

        //create dataStream
        DataStreamSource<Person> sourceDS = env.fromElements(tom, jerry);
        //TODO dataStream to table
        Table table = tEnv.fromDataStream(sourceDS);
        //TODO register table
        tEnv.createTemporaryView("SourceTable", table);
        Table tableResult = tEnv.sqlQuery("select * from SourceTable");

        //TODO table to dataStream
        DataStream<Row> rowDataStream = tEnv.toDataStream(tableResult);

        rowDataStream.print();

        env.execute();

    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
class Person{
    public String name;
    public Integer age;
}
